package com.ctyun.dws

import com.shujia.utils.SparkTool
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DwsScenicTouristMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 读取位置数据融合表
    val mergeLocDF: DataFrame = spark.table("dwi.dwi_res_regn_mergelocation_msk_d").where($"day_id" === day_id)
    // 读取景区配置表
    val scenicGridDF: DataFrame =
      spark.table("dim.dim_scenic_grid")
        // 将每个景区的多个网格展开
        .select($"scenic_id", $"scenic_name", $"grids", explode(split($"grids", ",")) as "grid_id")
    // 读取用户画像表
    val userTagDF: DataFrame = spark.table("dim.dim_usertag_msk_d").where($"day_id" === day_id)

    /**
     * 景区游客的定义：
     * 1、目的地在景区内
     * 2、停留时间大于30分钟
     * 3、居住地不在景区内
     */

    mergeLocDF
      // inner join, 直接通过grid_id关联，如果关联上 就说明在某个景区内
      .join(scenicGridDF.hint("broadcast"), "grid_id")
      // 停留时间大于30分钟
      .withColumn("min_start_date", min($"start_date") over Window.partitionBy($"mdn", $"scenic_id"))
      .withColumn("max_end_date", max($"end_date") over Window.partitionBy($"mdn", $"scenic_id"))
      .withColumn("stay_time", dateDiff($"min_start_date", $"max_end_date"))
      .where($"stay_time" > 30 * 60)
      // 关联用户画像表
      .join(userTagDF, "mdn")
      // 居住地不在景区内
      //      .where($"resi_grid_id" =!= $"grid_id")
      .where(!array_contains(split($"grids", ","), $"resi_grid_id"))
      .select($"mdn"
        , $"resi_county_id" as "source_county_id"
        , $"scenic_id" as "d_scenic_id"
        , $"scenic_name" as "d_scenic_name"
        , $"min_start_date" as "d_arrive_time"
        , round($"stay_time" / 3600, 2) as "d_stay_time"
      )
      .distinct()
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_scenic_tourist_msk_d/day_id=$day_id")

    // 增加分区
    spark.sql(
      s"""
         |alter table dws.dws_scenic_tourist_msk_d  add if not exists partition(day_id='$day_id')
         |""".stripMargin)


  }
}
